package com.rxjava.simplezip;

/**
 * 被观察者，用来生产发送事件；
 */
public class Observable<T> {

    private OnSubscribe<T> onSubscribe;

    private Observable(OnSubscribe<T> onSubscribe) {
    	//onSubscribe -> subscriber.onNext(7);
    	//onSubscribe -> subscriber.onNext("A");
    	//onSubscribe -> subscriber.onNext(new Observable<?>[]{o1, o2});
    	//onSubscribe -> OnSubscribeLift对象
        this.onSubscribe = onSubscribe;
    }

    public final void subscribe(Subscriber<? super T> subscriber) {
    	System.out.println("step1");
    	//subscriber -> System.out.println(s);
    	//subscriber -> queue.offer(o); tick();
    	//subscriber -> queue.offer(o); tick();两个
        onSubscribe.call(subscriber);
    }

    public static <T> Observable<T> create(OnSubscribe<T> onSubscribe) {
        return new Observable<>(onSubscribe);
    }

    public static <T1, T2, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, final Func2<? super T1, ? super T2, ? extends R> zipFunction) {
        return just(new Observable<?>[]{o1, o2}).lift(new OperatorZip<R>(zipFunction));
    }

    public static <T> Observable<T> just(final T value) {
        return create(new OnSubscribe<T>() {
            @Override
            public void call(Subscriber<? super T> subscriber) {
                subscriber.onNext(value);
            }
        });
    }

    public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
    	//onSubscribe -> subscriber.onNext(new Observable<?>[]{o1, o2});
    	//operator -> OperatorZip对象
        return new Observable<>(new OnSubscribeLift<T, R>(onSubscribe, operator));
    }

}